package com.ybc.reactor;

import java.util.concurrent.Flow;

public class FlowDemo {

    public static void main(String[] args) {
        // 创建 Publisher
        Flow.Publisher<Integer> publisher = new MyPublisher();

        // 创建 Subscriber
        Flow.Subscriber<Integer> subscriber = new MySubscriber();

        // 发布给订阅者
        publisher.subscribe(subscriber);
    }

    // 自定义 Publisher 类
    static class MyPublisher implements Flow.Publisher<Integer> {

        @Override
        public void subscribe(Flow.Subscriber<? super Integer> subscriber) {
            // 创建并传递 Subscription
            MySubscription subscription = new MySubscription(subscriber);
            subscriber.onSubscribe(subscription); // 通知订阅者已订阅
        }

        // 这个 Publisher 负责生成数据
        static class MySubscription implements Flow.Subscription {
            private final Flow.Subscriber<? super Integer> subscriber;
            private boolean isCancelled = false;
            private long requested = 0; // 控制请求的数量

            public MySubscription(Flow.Subscriber<? super Integer> subscriber) {
                this.subscriber = subscriber;
            }

            @Override
            public void request(long n) {
                if (n <= 0) {
                    throw new IllegalArgumentException("Request amount must be positive");
                }

                // 如果没有被取消，继续发送数据
                if (!isCancelled) {
                    // 请求的数据量超过一个上限，限制发送
                    if (requested + n > 10) {
                        n = 10 - requested; // 限制最多请求 10 个数据
                    }

                    // 记录请求的数据量
                    requested += n;

                    // 发送数据
                    for (int data = 0; data < n; data++) {
                        if (isCancelled) {
                            return; // 如果被取消了，停止发送数据
                        }
                        subscriber.onNext(data);  // 向 Subscriber 发出数据
                    }

                    // 如果已经满足了请求的数量，调用 onComplete
                    if (requested >= 10) {
                        subscriber.onComplete();  // 数据发送完毕，调用 onComplete
                    }
                }
            }

            @Override
            public void cancel() {
                System.out.println("Subscription cancelled.");
                isCancelled = true;
            }
        }
    }

    // 自定义 Subscriber 类
    static class MySubscriber implements Flow.Subscriber<Integer> {
        private Flow.Subscription subscription;
        private int count = 0; // 控制请求数据的数量

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            this.subscription = subscription;
            // 初始请求数据量
            subscription.request(3); // 请求 5 个数据
        }

        @Override
        public void onNext(Integer item) {
            // 处理收到的数据
            System.out.println("Received: " + item);
            count++;

            // 每接收到一定数量的数据后再请求更多数据
            if (count % 3 == 0) {  // 每接收到 3 个数据时，再请求更多
                subscription.request(3);  // 请求更多数据
            }
        }

        @Override
        public void onError(Throwable throwable) {
            throwable.printStackTrace();
        }

        @Override
        public void onComplete() {
            System.out.println("Data flow complete");
        }
    }
}
